package com.atguigu.day01;

import com.atguigu.utils.ClickEvent;
import com.atguigu.utils.ClickSource;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

public class Example9 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .addSink(new MyJDBC());

        env.execute();
    }

    public static class MyJDBC extends RichSinkFunction<ClickEvent> {
        private Connection conn;
        private PreparedStatement insertStmt;
        private PreparedStatement updateStmt;
        @Override
        public void open(Configuration parameters) throws Exception {
            conn = DriverManager.getConnection(
                    "jdbc:mysql://localhost:3306/userbehavior?useSSL=false",
                    "root",
                    "root"
            );
            insertStmt = conn.prepareStatement("INSERT INTO clicks (username, url) VALUES (?, ?)");
            updateStmt = conn.prepareStatement("UPDATE clicks SET url = ? WHERE username = ?");
        }

        @Override
        public void invoke(ClickEvent value, Context context) throws Exception {
            updateStmt.setString(1, value.url);
            updateStmt.setString(2, value.username);
            updateStmt.execute();

            if (updateStmt.getUpdateCount() == 0) {
                insertStmt.setString(1, value.username);
                insertStmt.setString(2, value.url);
                insertStmt.execute();
            }
        }

        @Override
        public void close() throws Exception {
            insertStmt.close();
            updateStmt.close();
            conn.close();
        }
    }
}
